package com.zf.nospring.rocketmq.simple;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 测试 拉（Pull）模式 的消费。
 * 不推荐使用，因为其 提交 和 拉取 操作在同一个线程中，会造成提交延迟
 */
public class ConsumerPullTest {

    public static void main(String[] args) throws MQClientException {
        // 创建消息消费者, 指定消费者所属的组名。使用 Pull 模式
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("default-cg");
        // 指定Nameserver地址
        consumer.setNamesrvAddr("localhost:9876");

        // consumer.setMessageModel(MessageModel.BROADCASTING);

        // 指定消费者订阅的主题和标签
        consumer.subscribe("test_topic", "*");
        // 是否自动提交消息的offset，默认开启
        // 注意: 自动提交操作是在 poll 中完成的，也就是本次 poll 之前会判断是否当前时间是否到达提交时间，到达就提交否则就不提交。
        //     所以真正的提交时间间隔是和"poll的超时"相关的，超时时间越小，真正的提交间隔 越接近 设置的提交间隔。否则就越大于设置的提交时间
        consumer.setAutoCommit(true);
        // 默认的自动提交间隔时间是 5s，这里设置为 1s, 必须 >= 1s 才有效
        consumer.setAutoCommitIntervalMillis(1000);
        // 启动消息消费者
        consumer.start();

        // 循环拉取消息
        while (true) {
            // 从rocketmq拉取消息，默认最长等待5s，如果5s内没有获取到消息，则返回空列表
            // List<MessageExt> messages = consumer.poll();
            // 手动指定等待时间，单位毫秒
            List<MessageExt> messages = consumer.poll(1000);

            for (MessageExt message : messages) {
                System.out.println(new String(message.getBody()));
            }
        }

    }
}
